#pragma once

#include "Requestor.hpp"
#include "RpcCaller.hpp"
#include "Rpc_Registry.hpp"
#include "Rpc_Topic.hpp"
#include "../communal/Dispatcher.hpp"

namespace xu
{
    namespace client
    {
        class RegistryClient
        {
        public:
            using Ptr = std::shared_ptr<RegistryClient>;

            RegistryClient(const std::string &ip, int port)
                : _requestor(std::make_shared<Requestor>()),
                  _provider(std::make_shared<Provider>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                auto cb = std::bind(&xu::client::Requestor::OnRequest, _requestor.get(), std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<BaseMessage>(xu::MType::RSP_SERVICE, cb);

                _client = xu::ModuoClientFactory::create(ip, port);
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client->SetMessageCallback(msg_cb);
                _client->Connect();
            }

            bool registryMethod(const std::string &method, const Address &host)
            {
                return _provider->RegistryMethod(_client->Connection(), method, host);
            }

            BaseConnection::Ptr GetConn()
            {
                return _client->Connection();
            }

        private:
            Requestor::Ptr _requestor;
            Provider::Ptr _provider;
            BaseClient::Ptr _client;
            Dispatcher::Ptr _dispatcher;
        };

        class DiscoveryClient
        {
        public:
            using Ptr = std::shared_ptr<DiscoveryClient>;

            DiscoveryClient(const std::string &ip, int port, Discoverer::OfflineCallback callback)
                : _requestor(std::make_shared<Requestor>()),
                  _discoverer(std::make_shared<Discoverer>(_requestor, callback)),
                  _dispatcher(std::make_shared<Dispatcher>())
            {
                auto cb_1 = std::bind(&xu::client::Requestor::OnRequest, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                // 注意！！！！
                _dispatcher->registerHandler<BaseMessage>(xu::MType::RSP_SERVICE, cb_1);

                auto cb_2 = std::bind(&Discoverer::onServiceRequest, _discoverer.get(), std::placeholders::_1, std::placeholders::_2);
                // 注意！！！！
                _dispatcher->registerHandler<ServiceRequest>(xu::MType::REQ_SERVICE, cb_2);

                _client = xu::ModuoClientFactory::create(ip, port);
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client->SetMessageCallback(msg_cb);
                _client->Connect();
            }
 
            // 向外提供的服务发现接口
            bool serviceDiscovery(const std::string &method, Address &host)
            {
                return _discoverer->serviceDiscovery(_client->Connection(), method, host);
            }

        private:
            Requestor::Ptr _requestor;
            Discoverer::Ptr _discoverer;
            BaseClient::Ptr _client;
            Dispatcher::Ptr _dispatcher;
        };

        class RpcClient
        {
        public:
            using Ptr = std::shared_ptr<RpcClient>;
            RpcClient(bool enableDiscovery, const std::string &ip, int port)
                : _enableDiscovery(enableDiscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _dispatcher(std::make_shared<Dispatcher>()),
                  _caller(std::make_shared<RpcCaller>(_requestor))
            {
                auto cb = std::bind(&xu::client::Requestor::OnRequest, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(xu::MType::RSP_RPC, cb);

                if (_enableDiscovery)
                {
                    auto del = std::bind(&RpcClient::DleClient, this, std::placeholders::_1);
                    _discovery = std::make_shared<DiscoveryClient>(ip, port, del);
                }
                else
                {
                    auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                    _client = xu::ModuoClientFactory::create(ip, port);
                    _client->SetMessageCallback(msg_cb);
                    _client->Connect();
                }
            }

            bool Call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                BaseClient::Ptr client = GetClient(method);
                if (client.get() != nullptr)
                {
                    return _caller->Call(client->Connection(), method, params, result);
                }
                return false;
            }
            bool Call(const std::string &method, const Json::Value &params, RpcCaller::Async &result)
            {
                BaseClient::Ptr client = GetClient(method);
                if (client.get() != nullptr)
                {
                    return _caller->Call(client->Connection(), method, params, result);
                }
                return false;
            }
            bool Call(const std::string &method, const Json::Value &params, const RpcCaller::Callback &cb)
            {
                BaseClient::Ptr client = GetClient(method);
                if (client.get() != nullptr)
                {
                    return _caller->Call(client->Connection(), method, params, cb);
                }
                return false;
            }

        private:
            BaseClient::Ptr NewClient(const Address &host)
            {
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                BaseClient::Ptr client = xu::ModuoClientFactory::create(host.first, host.second);
                client->SetMessageCallback(msg_cb);
                client->Connect();
                AddClient(host, client);
                LOG(LogLevel::DEBUG)<<"加入了一个复用连接";

                return client;
            }
            BaseClient::Ptr GetClient(const std::string &mt)
            {
                if (_enableDiscovery)
                {
                    Address res;
                    bool ret = _discovery->serviceDiscovery(mt, res);
                    if (ret == false)
                    {
                        return BaseClient::Ptr();
                    }

                    
                    BaseClient::Ptr client = GetClient(res);
                    
                    if (client.get() == nullptr)
                    {
                        client = NewClient(res);
                        LOG(LogLevel::DEBUG)<<"获取新连接";
                    }
                    return client;
                }
                else
                {
                    return _client;
                }
            }

            BaseClient::Ptr GetClient(const Address &host)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it != _rpc_clients.end())
                {
                    LOG(LogLevel::DEBUG)<<"找到了一个复用连接";
                    return it->second;
                }

                LOG(LogLevel::DEBUG)<<"没找到一个复用连接";
                return BaseClient::Ptr();
            }

            void AddClient(const Address &host, const BaseClient::Ptr &client)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _rpc_clients.insert(std::make_pair(host, client));
            }

            void DleClient(const Address &host)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it != _rpc_clients.end())
                {
                    it->second->ShutDown();
                    _rpc_clients.erase(host);
                    LOG(LogLevel::DEBUG)<<"删除成功";

                }    
            }

        private:
            struct AddressHash
            {
                size_t operator()(const Address &addr) const
                {
                    std::string str = addr.first + std::to_string(addr.second);
                    return std::hash<std::string>{}(str);
                }
            };

            bool _enableDiscovery;
            Requestor::Ptr _requestor;
            BaseClient::Ptr _client;
            Dispatcher::Ptr _dispatcher;
            RpcCaller::Ptr _caller;
            DiscoveryClient::Ptr _discovery;
            std::mutex _mutex;
            std::unordered_map<Address, BaseClient::Ptr, AddressHash> _rpc_clients;
        };


        
        class TopicClient
        {
        public:
            using Ptr = std::shared_ptr<TopicClient>;
            TopicClient(const std::string &ip, int port)
            :_requestor(std::make_shared<Requestor>()),
            _topicmanager(std::make_shared<TopicManager>(_requestor)),
            _dispatcher(std::make_shared<Dispatcher>())
            {
                auto cb = std::bind(&client::Requestor::OnRequest, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(xu::MType::RSP_TOPIC, cb);

                auto topic_cd = std::bind(&client::TopicManager::onPublish, _topicmanager.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<TopicRequest>(xu::MType::REQ_TOPIC, topic_cd);

                _client = xu::ModuoClientFactory::create(ip, port);
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client->SetMessageCallback(msg_cb);
                _client->Connect();
            }

            // 创建主题
            bool Create(const std::string &key)
            {
                return _topicmanager->Create(_client->Connection(),key);
            }

            // 删除主题
            bool Remove(const std::string &key)
            {
                return _topicmanager->Remove(_client->Connection(),key);
            }

            // 订阅主题
            bool Subscribe(const std::string &key, const TopicManager::SubCallback &scb)
            {
                return _topicmanager->Subscribe(_client->Connection(),key,scb);
            }

            // 取消订阅
            bool Cancel(const std::string &key)
            {
                return _topicmanager->Cancel(_client->Connection(),key);
            }

            // 发布消息
            bool Publish(const std::string &key,const std::string &msg)
            {
                return _topicmanager->Publish(_client->Connection(),key,msg);
            }

            //下线
            void Shutdown()
            {
                _client->ShutDown();
            }
        private:
            Requestor::Ptr _requestor;
            TopicManager::Ptr _topicmanager;
            BaseClient::Ptr _client;
            Dispatcher::Ptr _dispatcher;
        };

    }

}
